import org.apache.spark.{SparkConf, SparkContext}  
  
// 初始化Spark配置
val conf = new SparkConf().setAppName("SparkRDDOperations").setMaster("local[*]")  
val sc = new SparkContext(conf)  
  
// 1. 创建包含100个元素的RDD，将大于50的元素乘以2并筛选出大于100的元素  
val rdd1 = sc.parallelize(1 to 100)  
// 筛选大于50的元素，乘以2，再筛选大于100的
val filteredRDD = rdd1.filter(_ > 50).map(_ * 2).filter(_ > 100)  
println("1. 大于100的元素：")  
// 打印  
filteredRDD.collect().foreach(println)  
  
// 2. 创建包含10个元素的RDD，将所有单词转换为小写形式  
// 创建包含3个字符串的RDD  
val rdd2 = sc.parallelize(Seq("Hello World", "Spark RDD Operations", "Big Data"))  
// 将每个字符串转换为小写并拆分成单词  
val wordsRDD = rdd2.flatMap(_.toLowerCase.split(" "))  
println("\n2. 单词列表（小写形式）：")  
// 打印  
wordsRDD.collect().foreach(println)  
  
// 3. 创建包含5个元素的RDD，计算每个名字对应的分数之和  
val rdd3 = sc.parallelize(Seq(("Alice", 90), ("Bob", 80), ("Charlie", 70), ("Alice", 85), ("Bob", 95)))  
// 使用reduceByKey对每个名字对应的分数进行求和  
val scoresRDD = rdd3.reduceByKey(_ + _)  
println("\n3. 每个名字对应的分数之和：")  
// 打印  
scoresRDD.collect().foreach(println)  
  
// 4. 创建包含一些字符串的RDD，并按字符串长度分区   
val rdd4 = sc.parallelize(Seq("apple", "banana", "orange", "pear", "watermelon", "grape", "pineapple"))  
// 根据长度是否小于等于5进行分区，并应用HashPartitioner进行分区  
val partitionedRDD = rdd4.map(x => (x.length <= 5, x)).partitionBy(new org.apache.spark.HashPartitioner(2))  
println("\n4. 分区后的RDD：")  
// 打印  
partitionedRDD.collect().foreach(println)  
  
// 停止
sc.stop()